package com.example.flinkcourse.lesson3.analytics;

import com.example.flinkcourse.lesson3.model.Transaction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 交易分析器
 * 使用 DataStream API 进行交易数据分析
 */
public class TransactionAnalytics {
    private static final Logger LOG = LoggerFactory.getLogger(TransactionAnalytics.class);

    /**
     * 分析交易流
     * @param inputStream 输入交易流
     * @return 分析结果流
     */
    public static DataStream<TransactionStats> analyzeTransactions(DataStream<Transaction> inputStream) {
        LOG.info("Analyzing transactions with DataStream API...");
        
        return inputStream
            // 按类型分组
            .keyBy(Transaction::getType)
            
            // 窗口处理
            .window(SlidingEventTimeWindows.of(Time.minutes(10), Time.minutes(1)))
            
            // 统计分析
            .aggregate(new TransactionStatsAggregator())
            
            // 设置并行度
            .setParallelism(4);
    }
}

/**
 * 交易统计聚合器
 * 用于在窗口内统计交易数据
 */
class TransactionStatsAggregator implements AggregateFunction<Transaction, TransactionStats, TransactionStats> {
    @Override
    public TransactionStats createAccumulator() {
        return new TransactionStats();
    }

    @Override
    public TransactionStats add(Transaction transaction, TransactionStats accumulator) {
        // 更新统计结果
        accumulator.setType(transaction.getType());
        accumulator.setCount(accumulator.getCount() + 1);
        accumulator.setTotalAmount(accumulator.getTotalAmount() + transaction.getAmount());
        accumulator.setMaxAmount(Math.max(accumulator.getMaxAmount(), transaction.getAmount()));
        accumulator.setMinAmount(Math.min(accumulator.getMinAmount(), transaction.getAmount()));
        accumulator.setLastTimestamp(transaction.getTimestamp());
        return accumulator;
    }

    @Override
    public TransactionStats getResult(TransactionStats accumulator) {
        // 计算平均值
        if (accumulator.getCount() > 0) {
            accumulator.setAvgAmount(accumulator.getTotalAmount() / accumulator.getCount());
        }
        return accumulator;
    }

    @Override
    public TransactionStats merge(TransactionStats a, TransactionStats b) {
        // 合并两个统计结果
        TransactionStats merged = new TransactionStats();
        merged.setType(a.getType());
        merged.setCount(a.getCount() + b.getCount());
        merged.setTotalAmount(a.getTotalAmount() + b.getTotalAmount());
        merged.setMaxAmount(Math.max(a.getMaxAmount(), b.getMaxAmount()));
        merged.setMinAmount(Math.min(a.getMinAmount(), b.getMinAmount()));
        merged.setLastTimestamp(Math.max(a.getLastTimestamp(), b.getLastTimestamp()));
        return merged;
    }
}

/**
 * 交易统计数据模型
 */
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
class TransactionStats {
    private String type;
    private long count;
    private double totalAmount;
    private double avgAmount;
    private double maxAmount;
    private double minAmount;
    private long lastTimestamp;
} 